/**
* @program: src
*
* @description:rpc客户端
*
* @author: Mr.chen
*
* @create: 2020-11-05 17:40
**/
package client

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/afex/hystrix-go/hystrix"
	"github.com/gin-gonic/gin"
	"ydd-im/util/interceptor/grpcloginterceptor"
	"ydd-im/util/interceptor/jaegerinterceptor"

	// 支持多个拦截器插件
	grpc_middeware "github.com/grpc-ecosystem/go-grpc-middleware"
	"github.com/opentracing/opentracing-go"

	"go.uber.org/zap"
	"google.golang.org/grpc"
	"io/ioutil"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
	"ydd-im/service/api/define"
	"ydd-im/service/api/services/wsservice/pb"
	"ydd-im/service/api/tools/balance"
	"ydd-im/service/api/tools/response"
	"ydd-im/service/api/tools/util"
	"ydd-im/util/logger"
)

// 普通的grpc连接
func grpcConn(addr string) *grpc.ClientConn {
	conn, err := grpc.Dial(addr, grpc.WithInsecure())
	if err != nil {
		logger.GetLg().Error("did not connect",
			zap.Any("err",err),
		)
	}
	return conn
}
// 添加grpc拦截器，将链路追踪的ip写入上下文中
func createGrpcConn(serviceAddress string, c *gin.Context) *grpc.ClientConn {

	var conn *grpc.ClientConn
	var err error

	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond * 500)
	defer cancel()

	if true {
		tracer, _            := c.Get("Tracer")
		parentSpanContext, _ := c.Get("ParentSpanContext")
		conn, err = grpc.DialContext(
			ctx,
			serviceAddress,
			grpc.WithInsecure(),
			grpc.WithBlock(),
			grpc.WithUnaryInterceptor(
				grpc_middeware.ChainUnaryClient(
					jaegerinterceptor.ClientInterceptor(tracer.(opentracing.Tracer), parentSpanContext.(opentracing.SpanContext)),
					grpcloginterceptor.ClientInterceptor(),
				),
			),
		)
	} else {
		conn, err = grpc.DialContext(
			ctx,
			serviceAddress,
			grpc.WithInsecure(),
			grpc.WithBlock(),
			grpc.WithUnaryInterceptor(
				grpc_middeware.ChainUnaryClient(
					grpcloginterceptor.ClientInterceptor(),
				),
			),
		)
	}

	if err != nil {
		fmt.Println(serviceAddress, "grpc conn err:", err)
	}
	return conn
}
func grpcConnPlus(addr string) (conn *grpc.ClientConn,err error) {

	conn, err = grpc.Dial(addr, grpc.WithInsecure())
	if err != nil {
		logger.GetLg().Error("did not connect",
			zap.Any("err",err),
		)
		return nil,err
	}
	return conn,nil
	// 添加jagger连接

}

func SendRpcUser(addr string, messageId, sendUserId, clientId string, code int, message string,data response.MessageData) { // 发送rpc增加熔断器
	conn := grpcConn(addr)
	defer conn.Close()
	hystrix.ConfigureCommand("SendRpcUser", hystrix.CommandConfig{
		Timeout:               1000,
		MaxConcurrentRequests: 100,
		RequestVolumeThreshold:50,
		ErrorPercentThreshold: 25,
	})
	err := hystrix.Do("SendRpcUser", func() error {
		logger.GetLg().Info("发送到服务器",
			zap.String("host",define.LocalHost),
			zap.String("port",define.Port),
			zap.String("addr",addr),
			zap.String("clientId",clientId),
			zap.String("msg",message),
		)
		c :=  pb.NewCommonServiceClient(conn)
		_, err := c.SendToUser(context.Background(), &pb.SendToUserReq{
			MessageId:  messageId,
			SendUserId: sendUserId,
			ClientId:   clientId,
			Code:       int32(code),
			Message:    message,
			Data:    &pb.MessageModel{
				MessageLogId:int64(data.MessageLogId),
				ToUserId:int64(data.ToUserId),
				MessageType:int64(data.MessageType),
				Media:int64(data.Media),
				Content:data.Content,
				Pic:data.Pic,
				Url:data.Url,
				SendTime:data.SendTime,
				SendUsername:data.SendUsername,
				SendUserId:int64(data.SendUserId),
				Dstid:int64(data.Dstid),
			},
			MessageLogId: util.IntToint64(data.MessageLogId),
		})
		if err != nil {
			logger.GetLg().Error("failed to call",
				zap.Any("err",err),
			)
		}
		return nil
	},  func(e error) error {
		//服务降级
		return nil
	})
	if err !=nil {
		logger.GetLg().Error("failed to call",
			zap.Any("err",err),
		)
	}

}
func CloseRpcClient(addr string, clientId, systemId string) {
	conn := grpcConn(addr)
	defer conn.Close()
	logger.GetLg().Info("发送关闭连接到服务器",
		zap.String("host",define.LocalHost),
		zap.String("port",define.Port),
		zap.String("addr",addr),
		zap.String("clientId",clientId),
	)
	c := pb.NewCommonServiceClient(conn)
	_, err := c.CloseClient(context.Background(), &pb.CloseClientReq{
		SystemId: systemId,
		ClientId: clientId,
	})
	if err != nil {
		logger.GetLg().Error("failed to call",
			zap.Any("err",err),
		)

	}
}

//绑定分组
func SendRpcBindGroup(addr string, systemId string, groupName string, clientId string, userId string, extend string) {
	conn := grpcConn(addr)
	defer conn.Close()

	c := pb.NewCommonServiceClient(conn)
	_, err := c.BindGroup(context.Background(), &pb.BindGroupReq{
		SystemId:  systemId,
		GroupName: groupName,
		ClientId:  clientId,
		UserId:    userId,
		Extend:    extend,
	})
	if err != nil {
		logger.GetLg().Error("failed to call",
			zap.Any("err",err),
		)
	}
}

//发送分组消息
func SendGroupBroadcast(systemId string, messageId, sendUserId, groupName string, code int, message string, data *string) {
	define.ServerListLock.Lock()
	defer define.ServerListLock.Unlock()
	for _, addr := range define.ServerList {
		conn := grpcConn(addr)
		defer conn.Close()
		c := pb.NewCommonServiceClient(conn)
		_, err := c.Send2Group(context.Background(), &pb.Send2GroupReq{
			SystemId:   systemId,
			MessageId:  messageId,
			SendUserId: sendUserId,
			GroupName:  groupName,
			Code:       int32(code),
			Message:    message,
			Data:       *data,
		})
		if err != nil {
			logger.GetLg().Error("failed to call",
				zap.Any("err",err),
			)
		}
	}
}

//发送系统信息
func SendSystemBroadcast(systemId string, messageId, sendUserId string, code int, message string, data *string) {
	define.ServerListLock.Lock()
	defer define.ServerListLock.Unlock()
	for _, addr := range define.ServerList {
		conn := grpcConn(addr)
		defer conn.Close()

		c := pb.NewCommonServiceClient(conn)
		_, err := c.Send2System(context.Background(), &pb.Send2SystemReq{
			SystemId:   systemId,
			MessageId:  messageId,
			SendUserId: sendUserId,
			Code:       int32(code),
			Message:    message,
			Data:       *data,
		})
		if err != nil {
			logger.GetLg().Error("failed to call",
				zap.Any("err",err),
			)
		}
	}
}

func GetOnlineListBroadcast(systemId *string, groupName *string) (clientIdList []string) {
	define.ServerListLock.Lock()
	defer define.ServerListLock.Unlock()
	serverCount := len(define.ServerList)
	onlineListChan := make(chan []string, serverCount)
	var wg sync.WaitGroup

	wg.Add(serverCount)
	for _, addr := range define.ServerList {
		go func(addr string) {
			conn := grpcConn(addr)
			defer conn.Close()
			hystrix.ConfigureCommand("getOnlineList", hystrix.CommandConfig{
				Timeout:               1000,
				MaxConcurrentRequests: 100,
				RequestVolumeThreshold:50,
				ErrorPercentThreshold: 25,
			})

			err := hystrix.Do("getOnlineList", func() error {
				c := pb.NewCommonServiceClient(conn)
				response, err := c.GetGroupClients(context.Background(), &pb.GetGroupClientsReq{
					SystemId:  *systemId,
					GroupName: *groupName,
				})
				if err != nil {
					return err
				} else {
					onlineListChan <- response.List
				}
				return nil
			},  func(e error) error {
				//服务降级
				onlineListChan <- []string{"1"}
				return nil
			})
			if err !=nil {
				logger.GetLg().Error("failed to call",
					zap.Any("err",err),
				)
			}

			wg.Done()
		}(addr)
	}
	wg.Wait()
	for i := 1; i <= serverCount; i++ {
		list, ok := <-onlineListChan
		if ok {
			clientIdList = append(clientIdList, list...)
		} else {
			return
		}
	}

	close(onlineListChan)
	return
}
/**
 * 获取grpc地址
 * @access  public
 * @return  string addr ip+port
 * @return  error err
 */
func GetBalanceAddr() (addr string,err error) {
	if len(define.ServerList) ==0 {
		return "",errors.New("服务器出错")
	}

	var insts []*balance.Instance
	for k,_ :=range define.ServerList {
		ServerListAddr := strings.Split(define.ServerList[k],":")
		port,_ :=strconv.Atoi(ServerListAddr[1])
		one := balance.NewInstance(ServerListAddr[0],port)
		insts = append(insts, one)
	}
	var balanceName = "roundrobin"
	inst, err := balance.DoBalance(balanceName, insts)
	if err != nil {
		return "",err
	}
	addr = fmt.Sprintf("%s:%d",inst.GetHost(),inst.GetPort())
	fmt.Println(addr)
	return addr,nil
}
/**
 * 获取grpc地址,包含其它
 * @access  public
 * @return  string addr ip+port
 * @return  error err
 */
func GetBalanceAddrPlus(servername string,iscontain bool) (addr string,err error) {

	if len(define.ServerList) ==0 {
		logger.GetLg().Info("服务不存在",
			zap.String("host",define.LocalHost),
			zap.String("port",define.Port),
			zap.Any("err",errors.New("api微服务不可用")),
		)
		return "",errors.New("api微服务不可用")
	}
	var insts []*balance.Instance
	if len(servername) > 0{ //根据名称获取对应的地址
		if servername == define.ETCD_SERVER_Account_LIST {
			if len(define.AccountServerList) ==0 {
				logger.GetLg().Info("服务不存在",
					zap.String("host",define.LocalHost),
					zap.String("port",define.Port),
					zap.Any("err",errors.New("账号微服务不可用")),
				)
				return "",errors.New("账号微服务不可用")
			}
			for k,_ :=range define.AccountServerList {
				ServerListAddr := strings.Split(define.AccountServerList[k],":")
				port,_ :=strconv.Atoi(ServerListAddr[1])
				one := balance.NewInstance(ServerListAddr[0],port)
				insts = append(insts, one)
			}
		}else if servername == define.ETCD_SERVER_Group_LIST {
			if len(define.GroupServerList) ==0 {
				logger.GetLg().Info("服务不存在",
					zap.String("host",define.LocalHost),
					zap.String("port",define.Port),
					zap.Any("err",errors.New("群组微服务不可用")),
				)
				return "",errors.New("群组微服务不可用")
			}
			for k,_ :=range define.GroupServerList {
				ServerListAddr := strings.Split(define.GroupServerList[k],":")
				port,_ :=strconv.Atoi(ServerListAddr[1])
				one := balance.NewInstance(ServerListAddr[0],port)
				insts = append(insts, one)
			}
		}
	}else{
		for k,_ :=range define.ServerList {
			ServerListAddr := strings.Split(define.ServerList[k],":")
			port,_ :=strconv.Atoi(ServerListAddr[1])
			one := balance.NewInstance(ServerListAddr[0],port)
			insts = append(insts, one)
		}
	}
	if len(insts)==0 {
		logger.GetLg().Info("服务不存在",
			zap.String("host",define.LocalHost),
			zap.String("port",define.Port),
			zap.Any("err",errors.New("远程地址不能为空")),
		)
		return "",errors.New("远程地址不能为空")
	}
	var balanceName = "roundrobin"
	inst, err := balance.DoBalance(balanceName, insts)
	if err != nil {
		return "",err
	}
	addr = fmt.Sprintf("%s:%d",inst.GetHost(),inst.GetPort())
	return addr,nil
}
/**
 *登录
 * @access  public
 * @param   string     username  账号
 * @param   string     passwd  密码
 * @param   string     client  客户端
 * @param   string     appkey  appkey
 * @return  map[string]interface{} data 用户信息
 * @return  error err
 */
func LoginIsCluster01(username,passwd,client,appkey string) (data interface{},err error){
	addr,err := GetBalanceAddr()
	data = make(map[string]interface{},0)
	if err != nil {
		return data,err
	}
	conn,err := grpcConnPlus(addr)
	defer conn.Close()
	if err != nil {
		return data,err
	}
	hystrix.ConfigureCommand("Login", hystrix.CommandConfig{
		Timeout:               1000,
		MaxConcurrentRequests: 100,
		RequestVolumeThreshold:50,
		ErrorPercentThreshold: 25,
	})
	dataChan := make(chan *pb.LoginModel, 1)
	err = hystrix.Do("Login", func() error {
		c :=  pb.NewUserServiceClient(conn)
		resp,err := c.Login(context.Background(), &pb.LoginReq{
			Username:username,
			Passwd:passwd,
			Client:client,
			Appkey:appkey,
		})
		if err != nil  {
			return err
		}
		if int(resp.Code)!=int(response.SuccessCode) {
			return errors.New(resp.Msg)
		}
		dataChan<-resp.Data
		return nil
	}, nil) // fallback没有处理错误请置为空，否则会处理错误
	if err !=nil {
		logger.GetLg().Error("failed to call",
			zap.Any("err",err),
		)
		return data,err
	}

	if list, ok := <-dataChan;ok {
		close(dataChan)
		return list,nil
	}
	close(dataChan)
	return data,nil


}

/**
 *注册账号
 * @access  public
 * @param   string     username  账号
 * @param   string     passwd  密码
 * @param   string     userkey  用户id
 * @param   string     appkey  appkey
 * @return  mix
 */
func RegisterIsCluster(username,passwd,userkey,appkey string) (*define.ExecResult, error) {
	uInfo, _ := json.Marshal([]interface{}{username, passwd,userkey,appkey})
	res, err := execAction("/user/register", uInfo,define.ETCD_SERVER_Account_LIST,true)
	return parseBody(res), err
}
/**
 *登录
 * @access  public
 * @param   string     username  账号
 * @param   string     passwd  密码
 * @param   string     client  客户端
 * @param   string     appkey  appkey
 * @return  map[string]interface{} data 用户信息
 * @return  error err
 */
func LoginIsCluster(username,passwd,client,appkey string) (*define.ExecResult, error) {
	uInfo, _ := json.Marshal([]interface{}{username, passwd,client,appkey})
	res, err := execAction("/user/login", uInfo,define.ETCD_SERVER_Account_LIST,true)
	return parseBody(res), err
}

/**
 *rpc调用函数封装execAction : 向dbproxy请求执行action
 * @access  public
 * @param   string     funcName  函数名
 * @param   string     paramJson  参数
 * @return *pb.RespExec data 返回结果
 * @return  error err
 */
func execAction(funcName string, paramJson []byte,servername string,iscontain bool) (data *pb.RespExec,err  error) {
	addr,err := GetBalanceAddrPlus(servername,iscontain)
	if err != nil {
		return nil,err
	}
	fmt.Println(addr)
	conn,err := grpcConnPlus(addr)
	defer conn.Close()
	if err != nil {
		return nil,err
	}
	hystrix.ConfigureCommand("Login", hystrix.CommandConfig{
		Timeout:               1000,
		MaxConcurrentRequests: 100,
		RequestVolumeThreshold:50,
		ErrorPercentThreshold: 25,
	})
	dataChan := make(chan *pb.RespExec, 1)
	err = hystrix.Do("Login", func() error {
		dbCli :=  pb.NewDBProxyServiceClient(conn)
		resp,err := dbCli.ExecuteAction(context.TODO(), &pb.ReqExec{
			Action: []*pb.SingleAction{
				&pb.SingleAction{
					Name:   funcName,
					Params: paramJson,
				},
			},
		})

		if err != nil  {
			return err
		}
		dataChan<-resp
		return nil
	}, nil) // fallback没有处理错误请置为空，否则会处理错误
	if err !=nil {
		logger.GetLg().Error("failed to call",
			zap.Any("err",err),
		)
		return data,err
	}
	if list, ok := <-dataChan;ok {
		close(dataChan)
		return list,nil
	}
	close(dataChan)
	return data,nil

}

// parseBody : 转换rpc返回的结果
func parseBody(resp *pb.RespExec) *define.ExecResult {
	if resp == nil || resp.Data == nil {
		return nil
	}
	resList := []define.ExecResult{}
	_ = json.Unmarshal(resp.Data, &resList)
	// TODO:
	if len(resList) > 0 {
		return &resList[0]
	}
	return nil
}
/**
 *注册账号
 * @access  public
 * @param   string     username  账号
 * @param   string     passwd  密码
 * @param   string     userkey  用户id
 * @param   string     appkey  appkey
 * @return  mix
 */
func RegisterIsCluster01(username,passwd,userkey,appkey string) (err error){
	addr,err := GetBalanceAddr()
	if err != nil {
		return err
	}
	conn,err := grpcConnPlus(addr)
	defer conn.Close()
	if err != nil {
		return err
	}
	//c :=  pb.NewUserServiceClient(conn)
	//resp,err := c.Register(context.Background(), &pb.RegisterReq{
	//	Username:username,
	//	Passwd:passwd,
	//	Userkey:userkey,
	//	Appkey:appkey,
	//})

	hystrix.ConfigureCommand("Register", hystrix.CommandConfig{
		Timeout:               1000,
		MaxConcurrentRequests: 100,
		RequestVolumeThreshold:50,
		ErrorPercentThreshold: 25,
	})
	err = hystrix.Do("Register", func() error {
		c :=  pb.NewUserServiceClient(conn)
		resp,err := c.Register(context.Background(), &pb.RegisterReq{
			Username:username,
			Passwd:passwd,
			Userkey:userkey,
			Appkey:appkey,
		})
		if err != nil  {
			return err
		}
		if int(resp.Code)!=int(response.SuccessCode) {
			return errors.New(resp.Msg)
		}
		return nil
	}, nil) // fallback没有处理错误请置为空，否则会处理错误
	if err !=nil {
		logger.GetLg().Error("failed to call",
			zap.Any("err",err),
		)
		return err
	}
	return nil
}

// 前提启动之前多个后端服务
// 基本方式调用后端服务
func callAPI(addr, path, method string) (string, error) {
	req, _ := http.NewRequest(method, "http://"+addr+path, nil)
	client := http.DefaultClient
	res, err := client.Do(req)
	if err != nil {
		return "", err
	}
	defer res.Body.Close()
	buf, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return "", err
	}
	return string(buf), nil
}
func GetServiceAddr(serviceName string)(address string){
	return ""

}